package com.xzx.mr.mapper;

import com.xzx.mr.kv.AnalysisKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

/**
 * @author xinzhixuan
 * @version 1.0
 * @date 2024/8/10 18:08
 */
@Slf4j
public class MyHBaseMapper extends TableMapper<AnalysisKey, LongWritable> {
    private static final byte[] family = "d".getBytes();


    @Override
    protected void map(ImmutableBytesWritable key, Result value, Mapper<ImmutableBytesWritable, Result, AnalysisKey, LongWritable>.Context context) throws IOException, InterruptedException {
        String caller = Bytes.toString(value.getValue(family, "caller".getBytes()));
        String callee = Bytes.toString(value.getValue(family, "callee".getBytes()));
        String callTs = Bytes.toString(value.getValue(family, "callTs".getBytes()));
        int duration = Bytes.toInt(value.getValue(family, "duration".getBytes()));
        log.info("==================caller:{}", caller);
        //主叫 天
        context.write(AnalysisKey.of(caller, callTs.substring(0, 8)), new LongWritable(duration));
        //主叫 月
        context.write(AnalysisKey.of(caller, callTs.substring(0, 6)), new LongWritable(duration));
        //主叫 年
        context.write(AnalysisKey.of(caller, callTs.substring(0, 4)), new LongWritable(duration));

        //备叫 天
        context.write(AnalysisKey.of(callee, callTs.substring(0, 8)), new LongWritable(duration));
        //备叫 月
        context.write(AnalysisKey.of(callee, callTs.substring(0, 6)), new LongWritable(duration));
        //备叫 年
        context.write(AnalysisKey.of(callee, callTs.substring(0, 4)), new LongWritable(duration));

    }
}
